package br.com.felinelayer.common;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 */
public final class Communication implements Runnable {

    private static final SimpleDateFormat formatter = new SimpleDateFormat( "hh:mm:ss:SSS" );
    private DataOutputStream dataOutputStream = null;
    private ICommunicationListener communicationListener;
    private int _inquireId = 0;
    private Map<Integer, Semaphore> waitingInquireLock = new HashMap<>();
    private Map<Integer, Message> waitingInquireResponse = new HashMap<>();
    private HashMap<Thread, ArrayList<Message>> _batch = new HashMap<>();
    private boolean inBatch = false;
    public boolean isServer = false;

    private synchronized Socket getSocket() {
        return communicationListener.getSocket();
    }

    private synchronized int incInquireId() {
        int r = ++_inquireId;
        return r;
    }

    private synchronized void putInquireLock( Integer id, Semaphore th ) {
        waitingInquireLock.put( id, th );
    }

    private synchronized Semaphore getInquireLock( Integer id ) {
        return waitingInquireLock.get( id );
    }

    private synchronized Message getInquireResponse( Integer id ) {
        return waitingInquireResponse.get( id );
    }

    private synchronized void putInquireResponse( Integer id, Message message ) {
        waitingInquireResponse.put( id, message );
    }

    private synchronized void removeInquireResponse( Integer id ) {
        waitingInquireResponse.remove( id );
    }

    private synchronized ArrayList<Message> batch() {
        ArrayList<Message> result = _batch.get( Thread.currentThread() );
        if ( result == null ) {
            result = new ArrayList<>();
            _batch.put( Thread.currentThread(), result );
        }
        return result;
    }

    private Message getMessageFromBatch( String name, int componentId ) {
        for ( Message c : batch() ) {
            if ( c.getIntByName( Protocol.COMPONENT_ID ) != componentId ) {
                continue;
            }
            String n = c.getName();
            if ( n.equals( name ) ) {
                return c;
            }
        }
        return null;
    }

    private Message getMessageFromBatch( String[] names, int componentId ) {
        for ( Message c : batch() ) {
            if ( c.getIntByName( Protocol.COMPONENT_ID ) != componentId ) {
                continue;
            }
            String[] ns = c.getNames();
            if ( Arrays.equals( ns, names ) ) {
                return c;
            }
        }
        return null;
    }

    private synchronized Communication addBatch( Message message ) {
        ArrayList<Message> batch = batch();
        if ( message.getName().equals( Protocol.MSG_NAME_COMPONENT_PROPERTY_CHANGE ) ) {
            int componentId = message.getIntByName( Protocol.COMPONENT_ID );
            // Check if exist message in que to some component and property then rewrite
            Message m = getMessageFromBatch( message.getNames(), componentId );
            if ( m != null ) {
                batch.remove( m );
            } else {
                // Check if exists message in queue to some component
                m = getMessageFromBatch( Protocol.MSG_NAME_CREATE_COMPONENT, componentId );
                if ( m == null ) {
                    m = getMessageFromBatch( Protocol.MSG_NAME_COMPONENT_PROPERTY_CHANGE,
                            componentId );
                }
                // If found message with property of some component then acumulates
                if ( m != null ) {
                    m.copyValuesFrom( message );
                    return this;
                }
            }
        }
        batch.add( message );
        return this;
    }

    private Message markTimeSent( Message message ) {
        if ( message.isPropertyExists( Protocol.MSG_TIME_SENT ) ) {
            message.setValueByName( Protocol.MSG_TIME_SOURCE_SENT, message.getLongByName(
                    Protocol.MSG_TIME_SENT ) );
        }
        message.setValueByName( Protocol.MSG_TIME_SENT, System.currentTimeMillis() );
        return message;
    }

    private String prefix() {
        if ( isServer ) {
            return "PROTOCOL SERVER: ";
        } else {
            return "PROTOCOL CLIENT: ";
        }
    }

    private String readText( DataInputStream dataInputStream ) throws IOException {
        return dataInputStream.readUTF();
    }

    private void writeText( DataOutputStream dataOutputStream, String message ) throws IOException {
        dataOutputStream.writeUTF( message );
        dataOutputStream.flush();
    }

    // inquire -> createInquire
    private Inquire createInquire() {
        Inquire inq = new Inquire();
        Semaphore thread = new Semaphore( 0 ); //Thread
        inq.thread = thread;
        inq.id = new Integer( incInquireId() );
        putInquireLock( inq.id, thread );
        return inq;
    }

    private class Inquire {

        public Semaphore thread;
        public Integer id;
    }

    // run -> new FireMessageReceived
    private class FireMessageReceived extends Thread {

        private Communication communication;
        private ICommunicationListener communicationListener;
        private String text;

        public void fireMessageReceived( Communication communication,
                ICommunicationListener communicationListener, String text ) {
            if ( (text == null) || text.isEmpty() ) {
                return;
            }
            this.text = text;
            this.communication = communication;
            this.communicationListener = communicationListener;
            start();
        }

        @Override
        public void run() {
            // Convert text message do protocol object
            Message message = new Message().setFromString( text );
//            FelLogger.log( prefix() + " Received : " + text );
            // Define thread name, to help debug
            setName( message.toString() );
            // Process message received
            communicationListener.processMessageReceived( message );
            // If must wait inquire response
            communication.wakeUpInquiringThread( message );
        }
    }

    /**
     * run -> new FireMessageReceived FireMessageReceived ->
     * wakeUpInquiringThread
     *
     * responseInquire
     *
     * @param message
     */
    private synchronized void wakeUpInquiringThread( Message message ) {
        if ( !message.isPropertyExists( Protocol.INQUIRED_ID ) ) {
            return;
        }
        //
        String sId = message.getValueByName( Protocol.INQUIRED_ID );
        if ( sId.isEmpty() ) {
            return;
        }
        Integer id = new Integer( sId );
        Semaphore th = getInquireLock( id );
        if ( th != null ) {
            putInquireResponse( id, message );

            th.release();
            //System.out.println( "thread " + id + " acordei!" );
        } else {
            System.out.println( prefix() + "thread " + id
                    + " não encontrada! mensagem:" + message.toString() );
        }
    }

    public Communication( ICommunicationListener communicationListener ) {
        setCommunicationListener( communicationListener );
    }

    /**
     * Loop to listen from other side
     */
    @Override
    public void run() {
        try {
            DataInputStream input = new DataInputStream( getSocket().getInputStream() );
            // Loop to listen from other side
            while ( true ) {
                // If other side desconnected
                if ( (input == null) || (getSocket() == null) || (getSocket().isClosed()) ) {
                    System.out.println( prefix() + "peer desconnected!" );
                    break;
                }
                // Start thread to fire event from message received 
                new FireMessageReceived().fireMessageReceived( this,
                        communicationListener, readText( input ) );
            }
            if ( input != null ) {
                input.close();
            }
            dataOutputStream.close();
        } catch ( Exception ex ) {
            String m = ex.getMessage();
            if ( ((m != null) && (m.indexOf( "socket closed" ) >= 0)) || ((m != null)
                    && (m.indexOf( "Connection reset" ) >= 0))
                    || (ex instanceof java.io.EOFException) ) {
                System.err.println( prefix() + "peer desconectou!" );
            } else {
                Logger.getLogger( Communication.class.getName() ).log( Level.SEVERE, null, ex );
            }
        }
    }

    public void disconnect() {
        try {
            getSocket().close();
        } catch ( IOException ex ) {
            Logger.getLogger( Communication.class.getName() ).log( Level.SEVERE, null, ex );
        }
    }

    public void setCommunicationListener( ICommunicationListener communicationListener ) {
        try {
            this.communicationListener = communicationListener;
            this.dataOutputStream = new DataOutputStream( getSocket().getOutputStream() );
            System.out.println( prefix() + "Communication: waiting message from other side" );
            Thread thread = new Thread( this );
            thread.setName( "communication listener" );
            thread.start();
        } catch ( IOException ex ) {
            Logger.getLogger( Communication.class.getName() ).log( Level.SEVERE, null, ex );
        }
    }

    public Communication beginBatch() {
        inBatch = true;
        return this;
    }

    public boolean inBatch() {
        return inBatch;
    }

    public synchronized Communication endBatch() {
        ArrayList<Message> batch = batch();
        inBatch = false;
        if ( batch.size() > 0 ) {
            markTimeSent( batch.get( 0 ) );
            String fullString;
            if ( batch().size() == 1 ) {
                fullString = batch.get( 0 ).toString();
            } else {
                fullString = Message.arrayToMessageBatch( batch ).toString();
            }
            sendMessage( fullString );
            batch.clear();
        }
        return this;
    }

    public void finishMessageReceived( Message message ) {
        // Print a line to output data stream
        // FelLogger.log( prefix() + " Finish : " + message );
        // If need send inquire reply (confirmation)
        if ( message.isPropertyExists( Protocol.INQUIRING_ID ) ) {
            // Copy new object message
            message = message.cloneMessage();
            // Get id message wait inquiring
            int id = message.getIntByName( Protocol.INQUIRING_ID );
            message.deleteValue( Protocol.INQUIRING_ID );
            message.setValueByName( Protocol.INQUIRED_ID, id );
            sendMessage( message );
        }
    }

    public synchronized Communication sendMessage( Message message ) {
        if ( inBatch ) {
            addBatch( message );
            return this;
        }
        message = markTimeSent( message );
        sendMessage( message.toString() );
        return this;
    }

    //TODO: criar a thread somente para enviar? q cria um RunnAble que contém a mensagem, 
    // daí a thread singleton consome o runnable
    public synchronized void sendMessage( String message ) {
        try {
            // Create output data stream
            if ( communicationListener.getSocket().isClosed() ) {
                System.out.println( prefix() + "impossível enviar comando, cliente fechado" );
                return;
            }
            // Print a line to output data stream
            // FelLogger.log( prefix() + " Sent : " + message );
            // Send string to socket
            writeText( dataOutputStream, message );
        } catch ( IOException ex ) {
            Logger.getLogger( Communication.class.getName() ).log( Level.SEVERE,
                    "Exception senging message " + message, ex );
        }
    }

    public Message inquire( Message message ) {
        if ( inBatch ) {
            addBatch( message );
            return message;
        }
        try {
            Inquire inq = createInquire();
            message.setValueByName( Protocol.INQUIRING_ID, inq.id.toString() );
            sendMessage( message );
            inq.thread.acquire();
            message = getInquireResponse( inq.id );
            removeInquireResponse( inq.id );
            return message;
        } catch ( InterruptedException ex ) {
            Logger.getLogger( Communication.class.getName() ).log( Level.SEVERE, null, ex );
            return null;
        }
    }
}